package com.enzoism.springboot.kafka.controller;

import com.enzoism.springboot.kafka.common.AjaxResult;
import com.enzoism.springboot.kafka.model.UserModel;
import com.enzoism.springboot.kafka.template.MyPartitionTemplate;
import org.apache.kafka.clients.admin.AdminClient;
import org.apache.kafka.clients.admin.DescribeTopicsResult;
import org.apache.kafka.clients.admin.NewTopic;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.web.bind.annotation.*;

import javax.annotation.Resource;
import java.util.Arrays;
import java.util.concurrent.ExecutionException;

/**
 * 测试控制器
 * PS:@RestController 注解： 该注解是 @Controller 和 @ResponseBody 注解的合体版
 */
@RestController
@RequestMapping("/kafka")
public class KafkaController {

    @Resource
    private KafkaTemplate<String, Object> kafkaTemplate;
    @Resource
    private MyPartitionTemplate myPartitionTemplate;

    @Autowired
    private AdminClient adminClient;

    @GetMapping("/message/save")
    public AjaxResult send(@RequestParam String message){
        kafkaTemplate.send("testTopic",message);
        return AjaxResult.success("kafka发送消息成功");
    }

    @PostMapping("/user/save")
    public AjaxResult saveUser(@RequestBody UserModel user){
        kafkaTemplate.send("userTopic",user);
        return AjaxResult.success("kafka发送消息成功");
    }

    @GetMapping("/message/save2")
    public AjaxResult send2(@RequestParam String message){
        myPartitionTemplate.getKafkaTemplate().send("testTopic",message);
        return AjaxResult.success("kafka发送消息成功");
    }

    /**
     * 创建topic
     * @param topicName
     * @return
     */
    @GetMapping("/topic/create")
    public AjaxResult createTopic(@RequestParam String topicName) {
        NewTopic topic = new NewTopic(topicName, 2, (short) 1);
        adminClient.createTopics(Arrays.asList(topic));
        return AjaxResult.success("kafka创建主题成功",topicName);
    }

    /**
     * 查询topic
     * @param topicName
     * @return
     */
    @GetMapping("/topic/query")
    public AjaxResult queryTopic(@RequestParam String topicName) {
        DescribeTopicsResult result = adminClient.describeTopics(Arrays.asList(topicName));
        StringBuffer sb = new StringBuffer("topic信息:");
        try {
            result.all().get().forEach((k,v)->sb.append("key").append(k).append(";v:").append(v));
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (ExecutionException e) {
            e.printStackTrace();
        }
        return AjaxResult.success("kafka查询主题成功",sb.toString());
    }


}